package com.neusoft.databus.core.config;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

import com.google.common.eventbus.AsyncEventBus;
import com.neusoft.bizcore.web.utils.id.BaseIdGenerator;
import com.neusoft.bizcore.web.utils.id.DefaultIdGenerator;
import com.neusoft.databus.common.bean.CommandBean;
import com.neusoft.databus.common.utils.DefaultThreadFactory;
import com.neusoft.databus.core.constant.DatabusCoreConstant;
import com.neusoft.databus.core.flume.FlumeAvroRpcServer;
import com.neusoft.databus.core.flume.MetricCollectionProcessor;
import com.neusoft.databus.core.job.BuildJobService;

/**
 * Data Bus Core配置类
 *
 * @author sunchf
 * @date 2018年12月21日 上午10:38:54
 */
@Configuration
public class DatabusCoreConfiguration {

    @Value("${databus.core.monitorDataDispatcherMinThreadSize: 1000}")
    private int monitorDataDispatcherMinThreadSize;
    @Value("${databus.core.monitorDataDispatcherMaxThreadSize: 3000}")
    private int monitorDataDispatcherMaxThreadSize;
    @Value("${databus.core.modelUpdateEventThreadSize: 1}")
    private int modelUpdateEventThreadSize;
    @Value("${databus.core.redis.subscribeChannel: channel:databus}")
    private String redisSubscribeChannel;
    @Value("${databus.core.flume.server.port: 6666}")
    private int flumeRpcServerPort;
    @Autowired
    private MetricCollectionProcessor metricCollectionProcessor;

    @Bean
    public BaseIdGenerator gatewayCodeGenerator() {
        return new DefaultIdGenerator(0, DatabusCoreConstant.GATEWAY_CODE_PREFIX);
    }

    @Bean
    public BaseIdGenerator resourceCodeGenerator() {
        return new DefaultIdGenerator(0, DatabusCoreConstant.RESOURCE_CODE_PREFIX);
    }

    @Bean
    public BaseIdGenerator metricGroupCodeGenerator() {
        return new DefaultIdGenerator(0, DatabusCoreConstant.METRIC_GROUP_CODE_PREFIX);
    }

    @Bean
    public BaseIdGenerator metricCodeGenerator() {
        return new DefaultIdGenerator(0, DatabusCoreConstant.METRIC_CODE_PREFIX);
    }

    @Bean
    public AsyncEventBus monitorDataEventBus() {

        final ExecutorService executorService = new ThreadPoolExecutor(
                Integer.valueOf(this.monitorDataDispatcherMinThreadSize),
                Integer.valueOf(this.monitorDataDispatcherMaxThreadSize),
                120L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(),
                new DefaultThreadFactory(DatabusCoreConstant.MONITOR_DATA_EVENTBUS_ID),
                new ThreadPoolExecutor.CallerRunsPolicy());

        return new AsyncEventBus(DatabusCoreConstant.MONITOR_DATA_EVENTBUS_ID, executorService);
    }

    @Bean
    public AsyncEventBus modelUpdateEventBus() {
        return new AsyncEventBus(DatabusCoreConstant.MODEL_UPDATE_EVENTBUS_ID,
                Executors.newFixedThreadPool(this.modelUpdateEventThreadSize,
                        new DefaultThreadFactory(DatabusCoreConstant.MODEL_UPDATE_EVENTBUS_ID)));
    }

    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
            MessageListenerAdapter listenerAdapter) {

        final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(listenerAdapter, new PatternTopic(this.redisSubscribeChannel));

        return container;
    }

    @Bean
    MessageListenerAdapter listenerAdapter(BuildJobService listener) {
        return new MessageListenerAdapter(listener, "receiveRedisMessage");
    }

    @Bean
    @ConditionalOnMissingBean
    StringRedisTemplate template(RedisConnectionFactory connectionFactory) {
        return new StringRedisTemplate(connectionFactory);
    }

    @Bean
    @ConditionalOnMissingBean
    FlumeAvroRpcServer<CommandBean> flumeServer() {
        return new FlumeAvroRpcServer<>(this.flumeRpcServerPort, this.metricCollectionProcessor, CommandBean.class);
    }
}
